package cc.verywell.pureblock.msg.nfs.linkmap;

import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;


/**
* 
* Copyright © 2019 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2019年6月19日 下午4:51:11
*/
public class EasyCompactGrade implements EasyCompact {

	final private int grade;
	final private List<EasyTableHDhead> headsFrom;
	final private List<EasyTableHDhead> headsTo;
	final private List<EasyTableHDhead> headsBound;
	final private AtomicBoolean mutex = new AtomicBoolean(false);
	final private LinkCell[] lastCompacted;
	public EasyCompactGrade(int grade , List<EasyTableHDhead> headsFrom, List<EasyTableHDhead> headsTo, List<EasyTableHDhead> headsBound,LinkCell[] lastCompacted) {
		this.grade = grade;
		this.headsFrom = headsFrom;
		this.headsTo = headsTo;
		this.headsBound = headsBound;
		this.lastCompacted = lastCompacted;
	}
	
	@Override
	public void compact(EasyFileController efc, EasyCompactFinish cf) {
		if(mutex.getAndSet(true))throw new EasyIOException("compact already running!");
		
		ArrayList<EasyTableHDhead> gen = new ArrayList<>();
		LinkCell max = null;
		if(headsTo.size()==0 && grade>0) {
			///只移动文件;
			for (EasyTableHDhead head : headsTo) {
				max = head.end;
				gen.add(head.clone(this.grade+1));
			}
			
		}else {
			///正常Comact
			EasyIteratorGroup it = new EasyIteratorGroup();
			if(!headsFrom.isEmpty()) {
				if(this.grade==0) {
					for (EasyTableHDhead head : headsFrom) {
						it.add(head.getTable(efc.getAC()).get(LinkCell.MIN));
					}
				}else {
					it.add(new EasyTablesSorted.SortedHDIterator(headsFrom,0,efc.getAC(),LinkCell.MIN));
				}
			}
			if(!headsTo.isEmpty()) {
				it.add(new EasyTablesSorted.SortedHDIterator(headsTo,0,efc.getAC(),LinkCell.MIN));
			}
			
			CompactIterator cit = new CompactIterator(it,efc.getTableGroup(),headsBound);
			while(cit.hasNext()) {
				
				if(Thread.currentThread().isInterrupted())throw new EasyIOException("Interrupted");
				
				efc.runMemCompact();
				
				long newFileNumber = efc.getNewFileNumber();
				Path newFilePath = efc.getHDfile(newFileNumber);
				EasyTableHDhead head = EasyTableHD.add(this.grade+1, newFileNumber, newFilePath, cit, cit);
				gen.add(head);
			}
			max = cit.last();
			
		}
		
		synchronized (cf.sync()) {
			EasyTableGroup old = cf.syncGetGroup();
			
			EasyTablesUnsort unsort = old.getUnsort();
			EasyTablesSorted[] newSorteds = Arrays.copyOf(old.getSorteds(), old.getSorteds().length);
			if(this.grade==0) {
				HashSet<EasyTableHDhead> set = new HashSet<EasyTableHDhead>(unsort.files);
				set.removeAll(headsFrom);
				ArrayList<EasyTableHDhead> newL = new ArrayList<>(set);
				Collections.sort(newL);
				unsort = new EasyTablesUnsort(efc.getAC(),newL);
			}else {
				HashSet<EasyTableHDhead> set = new HashSet<EasyTableHDhead>(newSorteds[this.grade-1].files);
				set.removeAll(headsFrom);
				ArrayList<EasyTableHDhead> newL = new ArrayList<>(set);
				Collections.sort(newL);
				newSorteds[this.grade-1] = new EasyTablesSorted(efc.getAC(),newL);
			}
			{
				HashSet<EasyTableHDhead> set = new HashSet<EasyTableHDhead>(newSorteds[this.grade].files);
				set.removeAll(headsTo);
				ArrayList<EasyTableHDhead> newL = new ArrayList<>(set.size()+gen.size());
				newL.addAll(set);
				newL.addAll(gen);
				Collections.sort(newL);
				newSorteds[this.grade] = new EasyTablesSorted(efc.getAC(),newL);
			}
			EasyTableGroup ng = new EasyTableGroup(unsort, newSorteds);
			
			cf.syncLog().writetedCompact(headsFrom, headsTo, gen, efc.logGetFileNumber(), efc.logGetSeq());
			
			cf.syncSetGroup(ng);
		}
		
		lastCompacted[grade] = max;
		
		cf.doLast();
		
	}
	
	private class CompactIterator implements EasyIterator,EasyCompactSizeLimit{

		final EasyTableGroup old;
		final EasyIterator it;
		final List<EasyTableHDhead> headsBound;
		public CompactIterator(EasyIterator it,EasyTableGroup old,List<EasyTableHDhead> headsBound) {
			this.it = it;
			this.old = old;
			this.headsBound = headsBound;
			next();
		}

		private LinkCell current;

		@Override
		public boolean hasNext() {
			return current!=null;
		}
		@Override
		public LinkCell current() {
			return current;
		}
		
		private boolean fileStop = false; 
		@Override
		public void newFile() {
			fileStop = false;
		}
		@Override
		public boolean sizeLimit(long pos) {
			return fileStop || pos>EasyFileController.MAX_TABLE_SIZE;
		}
		
		private LinkCell otherLast;
		public LinkCell last() {
			return otherLast;
		}


		private int indexBound = 0;
		private long currentBound = 0;
		final private long BOUND_LIMIT = (long)(EasyFileController.MAX_TABLE_SIZE * EasyFileController.GRADE_EXPANDED);
		@Override
		public LinkCell next() {
			LinkCell cr = current;
			LinkCell last = cr; 
			current = null;
			while(it.hasNext()) {
				LinkCell nxt = it.next();
				if(!nxt.equalsWithoutSeq(last, 0)) {
					if(nxt.isRemove()) {
						LinkCell first = old.getFirst(grade+2, LinkCell.createNoSeq(nxt));
						if(first!=null && !first.isRemove()) {
							otherLast = current = nxt;
							break;
						}
						last = nxt;
					}else {
						otherLast = current = nxt;
						break;
					}
				}
			}
			
			//跨度太大要分开文件，以免下次compact时会使得资源占用过多;
			if(current!=null && !fileStop) {
				while(indexBound < headsBound.size() && current.compareTo(headsBound.get(indexBound).end)<=0) {
					currentBound += headsBound.get(indexBound).fileSize();
					indexBound++;
				}
				if(currentBound>BOUND_LIMIT) {
					currentBound = 0;
					fileStop = true;
				}
			}
			
			return cr;
		}

	}

	
	
}
